package com.java.spark

import org.apache.log4j.{Level, Logger}

import scala.Iterator
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession

object spark_allsuanzi {

  Logger.getLogger("org").setLevel(Level.ERROR)

  def main(args: Array[String]) {

    val spark = SparkSession.builder.master("local").appName("Word Count").getOrCreate();
    val sc = spark.sparkContext


//在Spark中创建RDD的创建方式大概可以分为三种：（1）、从集合中创建RDD；（2）、从外部存储创建RDD；（3）、从其他RDD创建。
    println("RDD创建方式 1")

    val file = sc.textFile("C:\\Users\\KC\\Desktop\\即饮咖啡.txt")

    println("RDD创建方式 2")
    val array = Array(1,2,3,4,5)

    val rdd = sc.parallelize(array)
    println( rdd.collect().foreach {print})

    val rdd1 = sc.makeRDD(array)  //make 自动分区
    println(rdd1.collect().foreach {print})
    //
    //    　　一、输入分区与输出分区一对一型
    //
    //    　　　　1、map算子
    //
    println("map算子")
    val maprdd = rdd.map(x => (x,1))
    println(maprdd.collect().foreach {print})


    //    　　　　2、flatMap算子
    //
    println("flatMap算子")

//    在spark中map函数和flatMap函数是两个比较常用的函数。其中
//    map：对集合中每个元素进行操作。
//    flatMap：对集合中每个元素进行操作然后再扁平化。
//    理解扁平化可以举个简单例子

    val arr=sc.parallelize(Array(("A",1),("B",2),("C",3)))
    arr.flatMap(x=>(x._1+x._2)).foreach(print(_,' '))

    println("flatMap算子应用")
    val arr1=sc.parallelize(Array("A;B;C;D;B;D;C","B;D;A;E;D;C","A;B"))
    //    　　　　3、mapPartitions算子

      arr1.map(_.split(";")).flatMap(x=>{
        for(i<-0 until x.length-1) yield (x(i)+","+x(i+1),1)
      }).reduceByKey(_+_).foreach(println)

    println("mapPartitions算子")
    def sumOfEveryPartition(input: Iterator[Int]): Int = {
      var total = 0
      input.foreach { elem =>
        total += elem
      }
      total
    }

    val input1 = sc.parallelize(List(1, 2, 3, 4, 5, 6), 2)//RDD有6个元素，分成2个partition
    val result1 = input1.mapPartitions(
      partition => Iterator(sumOfEveryPartition(partition)))//partition是传入的参数，是个list，要求返回也是list，即Iterator(sumOfEveryPartition(partition))
    result1.collect().foreach {
      println(_)
      //      # 6 15,分区计算和
    }


    //
    //    　　　　4、glom算子
    //
    println("glom算子")
//    一个RDD，glom之前为
//    在这里插入图片描述
//    glom之后在这里插入图片描述可见RDD中的元素已经变成了分片映射的列表
  val glomrdd = sc.parallelize(List(1, 2, 3, 4, 5, 6,7,8,9), 3)

    glomrdd.glom.foreach(array=>{
      println(array.mkString(","))
    })
    glomrdd.glom.foreach(array=>{
      println(array.toList)
    })

    //    　　二、输入分区与输出分区多对一型　
    //
    //    　　　　5、union算子
    println("union算子")
    //
    //    　　　　6、cartesian算子
    //
    println("cartesian算子 笛卡尔积")

    //    　　三、输入分区与输出分区多对多型
    //
    //    　　　　7、grouBy算子
    println("grouBy算子")
    //    　　四、输出分区为输入分区子集型
    val a = sc.parallelize(1 to 9, 3)
    a.groupBy(x => { if (x % 2 == 0) "even" else "odd" }).collect

    def myfunc(a: Int) : Int ={ a % 2 //分成两组
   }
    a.groupBy(myfunc).collect
    println("grouBykey算子")
    val a1 = sc.parallelize(List("dog", "tiger", "lion", "cat", "spider", "eagle"), 2)
    val b = a1.keyBy(_.length)//   给value加上key，key为对应string的长度
    b.groupByKey.collect
    //结果 Array((4,ArrayBuffer(lion)), (6,ArrayBuffer(spider)), (3,ArrayBuffer(dog, cat)), (5,ArrayBuffer(tiger, eagle)))


    //    　　　　8、filter算子
    println("filter算子")

    //    　　　　9、distinct算子
    println("distinct算子")
    //    　　　　10、subtract算子
    println("csubtract算子")
    //    　　　　11、sample算子
    println("sample算子")
    //    　　 12、takeSample算子
    println("takeSample算子")
    //    　　五、Cache型
    //
    //    　　　　13、cache算子　　
    println("cache算子　　")
    //    　　　　14、persist算子
    println("persist算子")
    // cache只有一个默认的缓存级别MEMORY_ONLY ，而persist可以根据情况设置其它的缓存级别。
    //
    //    2）Key-Value数据类型的Transfromation算子
    //
    //    　　一、输入分区与输出分区一对一
    //
    //    　　　　15、mapValues算子
    println("mapValues算子")
//    >>> x = sc.parallelize([("a", ["apple", "banana", "lemon"]), ("b", ["grapes"])])
//    >>> def f(x): return len(x)
//    >>> x.mapValues(f).collect()
 //     [('a', 3), ('b', 1)]

    //    　　　单个RDD聚集
    //
    //    　　　　16、combineByKey算子
    println("combineByKey算子")

    val initialScores = Array(("Fred", 88.0), ("Fred", 95.0), ("Fred", 91.0), ("Wilma", 93.0), ("Wilma", 95.0), ("Wilma", 98.0))
    val d1 = sc.parallelize(initialScores)
    type MVType = (Int, Double) //定义一个元组类型(科目计数器,分数)
    d1.combineByKey(
      score => (1, score),
      (c1: MVType, newScore) => (c1._1 + 1, c1._2 + newScore),
      (c1: MVType, c2: MVType) => (c1._1 + c2._1, c1._2 + c2._2)
    ).map { case (name, (num, socre)) => (name, socre / num) }.collect


    //    　　　　17、reduceByKey算子
    println("reduceByKey算子")

//
//    def reduceByKey(func: (V, V) => V): RDD[(K, V)]
//
//    def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD[(K, V)]
//
//    def reduceByKey(partitioner: Partitioner, func: (V, V) => V): RDD[(K, V)]
//
//    该函数用于将RDD[K,V]中每个K对应的V值根据映射函数来运算。
//
//    参数numPartitions用于指定分区数；
//
//    参数partitioner用于指定分区函数；
//    scala> var rdd1 = sc.makeRDD(Array(("A",0),("A",2),("B",1),("B",2),("C",1)))
//    rdd1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[91] at makeRDD at :21
//
//    scala> rdd1.partitions.size
//    res82: Int = 15
//
//    scala> var rdd2 = rdd1.reduceByKey((x,y) => x + y)
//    rdd2: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[94] at reduceByKey at :23
//
//    scala> rdd2.collect
//    res85: Array[(String, Int)] = Array((A,2), (B,3), (C,1))
//
//    scala> rdd2.partitions.size
//    res86: Int = 15
//
//    scala> var rdd2 = rdd1.reduceByKey(new org.apache.spark.HashPartitioner(2),(x,y) => x + y)
//    rdd2: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[95] at reduceByKey at :23
//
//    scala> rdd2.collect
//    res87: Array[(String, Int)] = Array((B,3), (A,2), (C,1))
//
//    scala> rdd2.partitions.size
//    res88: Int = 2
//




    val words = Array("one", "two", "two", "three", "three", "three")

    val wordPairsRDD = sc.parallelize(words).map(word => (word, 1))

    val wordCountsWithReduce = wordPairsRDD.reduceByKey(_ + _)

    val wordCountsWithGroup = wordPairsRDD.groupByKey().map(t => (t._1, t._2.sum))


    println("reduceByKey(_ + _)  == groupByKey().map(t => (t._1, t._2.sum))     ")








    //    　　　　18、partitionBy算子
    println("partitionBy算子")
    //    　　两个RDD聚集
//    该函数根据partitioner函数生成新的ShuffleRDD，将原RDD重新分区。
//    scala> var rdd1 = sc.makeRDD(Array((1,"A"),(2,"B"),(3,"C"),(4,"D")),2)
//    rdd1: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[23] at makeRDD at :21
//
//    scala> rdd1.partitions.size
//    res20: Int = 2
//
//    //查看rdd1中每个分区的元素
//    rdd1.mapPartitionsWithIndex{
//               (partIdx,iter) => {
//                   var part_map = scala.collection.mutable.Map[String,List[(Int,String)]]()
//                     while(iter.hasNext){
//                         var part_name = "part_" + partIdx;
//                         var elem = iter.next()
//                         if(part_map.contains(part_name)) {
//                             var elems = part_map(part_name)
//                             elems ::= elem
//                              part_map(part_name) = elems
//                           } else {
//                             part_map(part_name) = List[(Int,String)]{elem}
//                          }
//                       print(elem)
//                       }
//
//                     part_map.iterator
//
//                 }
//             }.collect
//    res22: Array[(String, List[(Int, String)])] = Array((part_0,List((2,B), (1,A))), (part_1,List((4,D), (3,C))))
//    //(2,B),(1,A)在part_0中，(4,D),(3,C)在part_1中
//
//    //使用partitionBy重分区
//    scala> var rdd2 = rdd1.partitionBy(new org.apache.spark.HashPartitioner(2))
//    rdd2: org.apache.spark.rdd.RDD[(Int, String)] = ShuffledRDD[25] at partitionBy at :23
//
//    scala> rdd2.partitions.size
//    res23: Int = 2
//
//    //查看rdd2中每个分区的元素
//    scala> rdd2.mapPartitionsWithIndex{
//      |         (partIdx,iter) => {
//        |           var part_map = scala.collection.mutable.Map[String,List[(Int,String)]]()
//        |             while(iter.hasNext){
//          |               var part_name = "part_" + partIdx;
//          |               var elem = iter.next()
//          |               if(part_map.contains(part_name)) {
//            |                 var elems = part_map(part_name)
//            |                 elems ::= elem
//              |                 part_map(part_name) = elems
//            |               } else {
//            |                 part_map(part_name) = List[(Int,String)]{elem}
//            |               }
//          |             }
//        |             part_map.iterator
//        |         }
//      |       }.collect
//    res24: Array[(String, List[(Int, String)])] = Array((part_0,List((4,D), (2,B))), (part_1,List((3,C), (1,A))))
    //(4,D),(2,B)在part_0中，(3,C),(1,A)在part_1中
    //    　　　　19、Cogroup算子
    println("Cogroup算子")
//
//    val sc = new SparkContext(sparkConf)
//
//    val rdd1 = sc.parallelize(Array(("aa",1),("bb",2),("cc",6)))
//
//    val rdd2 = sc.parallelize(Array(("aa",3),("dd",4),("aa",5)))
//
//    val rdd3 = rdd1.cogroup(rdd2).collect()
//
//    for (i <- 0 to rdd3.length-1){
//      println(rdd3(i))

    //    　　三、连接
    //
    //    　　　　20、join算子
    println("join算子")
//    var rdd1 = sc.makeRDD(Array(("A","1"),("B","2"),("C","3")),2)
//    var rdd2 = sc.makeRDD(Array(("A","a"),("C","c"),("D","d")),2)
//
//    scala> rdd1.join(rdd2).collect
//    res10: Array[(String, (String, String))] = Array((A,(1,a)), (C,(3,c)))
    //    　　　　21、leftOutJoin和 rightOutJoin算子
    println("leftOutJoin和 rightOutJoin算子")
//
//    spark 中join、leftOuterJoin以及rightOuterJoin的用法详解
//    这三个都是针对两个pair RDD的操作。语法为:
//      pair_rdd1.xxxx(pair_rdd2)，其中xxxx为三个转化操作。
//
//
//
//    1、join的用法
//    手动设定两个pair RDD
//      1
//    pairRDD1 = sc.parallelize([('Apple',10),('Orange',5),('pair',1)])
//    pairRDD2 = sc.parallelize([('Apple',5),('Orange',7),])
//
//    对pair RDD进行join操作：输出结果如下：
//
//    pairRDD1.join(pairRDD2).collect()
//    输出为：[('Orange', (5, 7)), ('Apple', (10, 5))]
//
//    pairRDD2.join(pairRDD1).collect()
//    输出为：[('Orange', (7, 5)), ('Apple', (5, 10))]
//
//    小结
//    join连接的两个RDD，只会输出具有相同键的值，值的排列顺序为转换xxxx操作左边RDD的值在前，右边RDD的值在后。
//
//    2、leftOuterJoin的用法
//    pairRDD1.leftOuterJoin(pairRDD2).collect()
//    输出为：[('Orange', (5, 7)), ('pair', (1, None)), ('Apple', (10, 5))]
//
//    pairRDD2.leftOuterJoin(pairRDD1).collect()
//    输出为：[('Orange', (7, 5)), ('Apple', (5, 10))]
//
//    小结
//    leftOuterJoin连接的两个RDD，以转换xxxx操作左边的RDD的键为基准，值的排列顺序为转换xxxx操作左边RDD的值在前，右边RDD的值在后，没有的值补充None。
//
//    3、rightOuterJoin的用法
//    pairRDD1.rightOuterJoin(pairRDD2).collect()
//    输出为：[('Orange', (5, 7)), ('Apple', (10, 5))]
//
//    pairRDD2.rightOuterJoin(pairRDD1).collect()
//    输出为：[('Orange', (7, 5)), ('pair', (None, 1)), ('Apple', (5, 10))]


    //
    //
    //
    //    3）Action算子
    //
    //    　　一、无输出
    //
    //    　　　　22、foreach算子
    println("foreach算子")
    //    　　二、HDFS
    //
    //    　　　　23、saveAsTextFile算子
    println("saveAsTextFile算子")
    //    　　　　24、saveAsObjectFile算子
    println("saveAsObjectFile算子")
    //    　　三、Scala集合和数据类型
    //
    //    　　　　25、collect算子
    println("collect算子")
    //    　　　　26、collectAsMap算子
    println("collectAsMap算子")

//    collectAsMap 的使用对象必须是Tuple 元组类型，在spark中将元组类型转换为Map类型，
//
//    应用示例：
//
//    val a = sc.parallelize(List(2,3,4,5))
//
//    val b = sc.parallelize(List("a","b","c","d"))
//
//    a.zip(b).collectAsMap 结果为：scala.collection.Map[Int,String] = Map(2 -> a, 5 -> d, 4 -> c, 3 -> b)
//
//

    //    　　　  27、reduceByKeyLocally算子
    println("reduceByKeyLocally算子")

//    reduceByKeyLocally
//    def reduceByKeyLocally(func: (V, V) => V): Map[K, V]
//
//    该函数将RDD[K,V]中每个K对应的V值根据映射函数来运算，运算结果映射到一个Map[K,V]中，而不是RDD[K,V]。
//    scala> var rdd1 = sc.makeRDD(Array(("A",0),("A",2),("B",1),("B",2),("C",1)))
//    rdd1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[91] at makeRDD at :21
//
//    scala> rdd1.reduceByKeyLocally((x,y) => x + y)
//    res90: scala.collection.Map[String,Int] = Map(B -> 3, A -> 2, C -> 1)

    //    　　　  28、lookup算子
    println("lookup算子子")
    //    　　　　29、count算子
    println("count算子")
    //    　　　　30、top算子
    println("top算子")
    //    　　　　31、reduce算子
    println("reduce算子")
    //    　　　　32、fold算子
    println("fold算子")
    //    　　　　33、aggregate算子
    println("aggregate算子")
    //
    //
//    aggregate
//
//    def aggregate[U](zeroValue: U)(seqOp: (U, T) ⇒ U, combOp: (U, U) ⇒ U)(implicit arg0: ClassTag[U]): U
//    aggregate用户聚合RDD中的元素，先使用seqOp将RDD中每个分区中的T类型元素聚合成U类型，再使用combOp将之前每个分区聚合后的U类型聚合成U类型，特别注意seqOp和combOp都会使用zeroValue的值，zeroValue的类型为U。
//    var rdd1 = sc.makeRDD(1 to 10,2)
//    rdd1.mapPartitionsWithIndex{
//      (partIdx,iter) => {
//        var part_map = scala.collection.mutable.Map[String,List[Int]]()
//        while(iter.hasNext){
//          var part_name = "part_" + partIdx;
//          var elem = iter.next()
//          if(part_map.contains(part_name)) {
//            var elems = part_map(part_name)
//            elems ::= elem
//            part_map(part_name) = elems
//          } else {
//            part_map(part_name) = List[Int]{elem}
//          }
//        }
//        part_map.iterator
//
//      }
//    }.collect
//    res16: Array[(String, List[Int])] = Array((part_0,List(5, 4, 3, 2, 1)), (part_1,List(10, 9, 8, 7, 6)))
//
//    ##第一个分区中包含5,4,3,2,1
//    ##第二个分区中包含10,9,8,7,6
//    scala> rdd1.aggregate(1)(
//      |           {(x : Int,y : Int) => x + y},
//      |           {(a : Int,b : Int) => a + b}
//        |     )
//    res17: Int = 58
//
//    结果为什么是58，看下面的计算过程：
//
//    ##先在每个分区中迭代执行 (x : Int,y : Int) => x + y 并且使用zeroValue的值1
//      ##即：part_0中 zeroValue+5+4+3+2+1 = 1+5+4+3+2+1 = 16
//    ## part_1中 zeroValue+10+9+8+7+6 = 1+10+9+8+7+6 = 41
//    ##再将两个分区的结果合并(a : Int,b : Int) => a + b ，并且使用zeroValue的值1
//    ##即：zeroValue+part_0+part_1 = 1 + 16 + 41 = 58
//
//    再比如：
//    scala> rdd1.aggregate(2)(
//      |           {(x : Int,y : Int) => x + y},
//      |           {(a : Int,b : Int) => a * b}
//        |     )
//    res18: Int = 1428
//
//    ##这次zeroValue=2
//    ##part_0中 zeroValue+5+4+3+2+1 = 2+5+4+3+2+1 = 17
//    ##part_1中 zeroValue+10+9+8+7+6 = 2+10+9+8+7+6 = 42
//    ##最后：zeroValue*part_0*part_1 = 2 * 17 * 42 = 1428
//
//    因此，zeroValue即确定了U的类型，也会对结果产生至关重要的影响，使用时候要特别注意。
//
//    fold
//
//    def fold(zeroValue: T)(op: (T, T) ⇒ T): T
//
//    fold是aggregate的简化，将aggregate中的seqOp和combOp使用同一个函数op。
//    scala> rdd1.fold(1)(
//      |       (x,y) => x + y
//    |     )
//    res19: Int = 58
//
//    ##结果同上面使用aggregate的第一个例子一样，即：
//    scala> rdd1.aggregate(1)(
//      |           {(x,y) => x + y},
//      |           {(a,b) => a + b}
//        |     )
//    res20: Int = 58
//
//    lookup
//
//    def lookup(key: K): Seq[V]
//
//    lookup用于(K,V)类型的RDD,指定K值，返回RDD中该K对应的所有V值。
//    scala> var rdd1 = sc.makeRDD(Array(("A",0),("A",2),("B",1),("B",2),("C",1)))
//    rdd1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[0] at makeRDD at :21
//
//    scala> rdd1.lookup("A")
//    res0: Seq[Int] = WrappedArray(0, 2)
//
//    scala> rdd1.lookup("B")
//    res1: Seq[Int] = WrappedArray(1, 2)



    spark.stop()
  }
}